package org.lzfto.flink.demo.business.flinkTest.flinkObjects.connector.file;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.Date;

import org.apache.flink.api.common.io.DefaultInputSplitAssigner;
import org.apache.flink.api.common.io.FileInputFormat;
import org.apache.flink.api.common.io.InputFormat;
import org.apache.flink.api.common.io.statistics.BaseStatistics;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.io.InputSplitAssigner;
import org.apache.flink.table.data.BoxedWrapperRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.poi.EncryptedDocumentException;
import org.apache.poi.ss.usermodel.Cell;
import org.apache.poi.ss.usermodel.CellType;
import org.apache.poi.ss.usermodel.DateUtil;
import org.apache.poi.ss.usermodel.Row;
import org.apache.poi.ss.usermodel.Sheet;
import org.apache.poi.ss.usermodel.Workbook;
import org.apache.poi.ss.usermodel.WorkbookFactory;
import org.lzfto.flink.demo.business.flinkTest.base.record.ExcelOptionRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class CustExcelInputFormat implements InputFormat<RowData, CustExcelSplit> {

	private final  Logger LOGGER = LoggerFactory.getLogger(getClass());
	private static final long serialVersionUID = 1L;

	private final ExcelOptionRecord option;
	private final DataType dataType;
	private Workbook workbook;
	private Sheet sheet;
	private int currentRow = 0;
	private int totalRow = 0;

	public CustExcelInputFormat(ExcelOptionRecord option, DataType dataType) {
		this.option = option;
		this.dataType = dataType;
		this.currentRow=0;
		this.totalRow=-1;
	}
	

	@Override
	public void configure(Configuration parameters) {
		LOGGER.info("打开自定义excel读写器");		
		//无意义练习，暂时不会影响什么东西
		parameters.setString("taskname", "lzfto的从excel读取数据到逻辑表的任务");
	}

	@Override
	public BaseStatistics getStatistics(BaseStatistics cachedStatistics) throws IOException {
		return new FileInputFormat.FileBaseStatistics(0, 100, BaseStatistics.AVG_RECORD_BYTES_UNKNOWN);
	}

	@Override
	public CustExcelSplit[] createInputSplits(int minNumSplits) throws IOException {
		// 此处需要实现，否则不会被调用
		// 把某个split的split数设置0，会导致某个split不会执行。
		// 因为本处仅仅是一个演示，所以实际只要一个执行即可。
		// 如果执行环境已经设置为并行度为1，那么可以考虑不那么处理。
		CustExcelSplit[] list = new CustExcelSplit[minNumSplits];
		for (int i = 0; i < minNumSplits; i++) {
			CustExcelSplit item = new CustExcelSplit(i+1);
			list[i] = item;
		}
		return list;
	}

	@Override
	public InputSplitAssigner getInputSplitAssigner(CustExcelSplit[] inputSplits) {
		return new DefaultInputSplitAssigner(inputSplits);
	}

	@Override
	public void open(CustExcelSplit split) throws IOException {
		//此数应该结合spit进行处理，避免重复操作
		LOGGER.info("开始执行open方法.分片序号:{}",split.getSplitNumber());
		try {
		    String fullPath=option.getFullPath(split.getSplitNumber());
			File targetFile = new File(fullPath);
			if(!targetFile.exists()) {
				LOGGER.warn("当前分片不存在对应文件:{}",targetFile);
				return;
			}
			else {
				LOGGER.info("当前分片对应文件:{}",targetFile);
			}
			FileInputStream file = new FileInputStream(targetFile);
			this.workbook = WorkbookFactory.create(file);
			this.sheet = workbook.getSheetAt(0); // 获取第一个sheet
			if (this.sheet == null) {
				throw new IOException("Excel 文件不包含任何工作表");
			} else {
				this.totalRow = this.sheet.getLastRowNum();
				this.currentRow = this.option.startRow();
				if (totalRow < 0)
					throw new IOException("工作表不包含任何行");
			}
		} catch (EncryptedDocumentException e) {
			throw new IOException("此excel文档本加密过，无法打开", e);
		}
	}

	@Override
	public boolean reachedEnd() throws IOException {
		return this.currentRow > this.totalRow;
	}

	@Override
	public RowData nextRecord(RowData reuse) throws IOException {
		if (sheet == null) {
			return null;
		}

		LOGGER.info("读取第{}行excel行", currentRow);
		org.apache.poi.ss.usermodel.Row row = sheet.getRow(currentRow);
		if (row == null) {
			currentRow++;
			return null;
		}

		int arity = dataType.getChildren().size();
		BoxedWrapperRowData rowData = new BoxedWrapperRowData(arity);

		for (int i = 0; i < arity; i++) {
			DataType fieldType = dataType.getChildren().get(i);
			LogicalTypeRoot type = fieldType.getLogicalType().getTypeRoot();
			String fieldTypeName = type.name();
			Cell cell = row.getCell(i, Row.MissingCellPolicy.RETURN_BLANK_AS_NULL);

			try {
				if (cell == null || cell.getCellType() == CellType.BLANK) {
					rowData.setNullAt(i);
					continue;
				}

				switch (type) {
				case TINYINT, SMALLINT, INTEGER -> {
					rowData.setInt(i, (int) cell.getNumericCellValue());
				}
				case BIGINT -> rowData.setLong(i, (long) cell.getNumericCellValue());
				case FLOAT -> rowData.setFloat(i, (float) cell.getNumericCellValue());
				case DOUBLE -> rowData.setDouble(i, cell.getNumericCellValue());
				case BOOLEAN -> {
					if (cell.getCellType() == CellType.BOOLEAN) {
						rowData.setBoolean(i, cell.getBooleanCellValue());
					} else {
						rowData.setBoolean(i, Boolean.parseBoolean(cell.toString()));
					}
				}
				case DATE -> {
					if (DateUtil.isCellDateFormatted(cell)) {
						Date tmpDate = cell.getDateCellValue();
						int days = (int) (tmpDate.getTime() / 86400 / 1000);
						rowData.setInt(i, days);
					} else {
						throw new IOException("excel单元不是日期格式");
					}
				}
				case TIMESTAMP_WITH_TIME_ZONE, TIMESTAMP_WITHOUT_TIME_ZONE, TIMESTAMP_WITH_LOCAL_TIME_ZONE -> {
					if (DateUtil.isCellDateFormatted(cell)) {
						Date tmpDate = cell.getDateCellValue();
						TimestampData date = TimestampData.fromEpochMillis(tmpDate.getTime());
						rowData.setTimestamp(i, date, 0);
					} else {
						throw new IOException("excel单元不是日期时间格式");
					}
				}
				case CHAR -> {
					if (cell.getCellType() == CellType.STRING) {
						String tmp = cell.getStringCellValue();
						if (tmp != null && tmp.length() == 1) {
							String strTmp = String.valueOf(tmp.charAt(0));
							StringData data = StringData.fromString(strTmp);
							rowData.setNonPrimitiveValue(i, data);
						} else {
							rowData.setNullAt(i);
						}
					}
				}
				case VARCHAR -> {
					if (cell.getCellType() == CellType.STRING) {
						String tmpStr = cell.getStringCellValue();
						StringData data = StringData.fromString(tmpStr);
						rowData.setNonPrimitiveValue(i, data);
					} else {
						rowData.setNullAt(i);
					}
				}
				default -> {
					String tmpStr = cell.getStringCellValue();
					StringData data = StringData.fromString(tmpStr == null ? "" : tmpStr);
					rowData.setNonPrimitiveValue(i, data);
				}
				}
			} catch (Exception e) {
				throw new IOException(String.format("读取单元[row= %d, column= %d]为类型%s",
						currentRow, i, fieldTypeName), e);
			}
		}
		currentRow++;
		return rowData;
	}

	@Override
	public void close() throws IOException {
		if (this.workbook != null) {
			this.workbook.close();
		}
	}

}
